package com.chukun.flink.stream.state.key;

import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ReducingState;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author chukun
 * @version 1.0.0
 * @description 带状态的flatmap操作符 ReducingStateDescriptor的基本使用
 * @createTime 2022年05月15日 22:59:00
 */
public class ReducingStateFlatMap extends RichFlatMapFunction<Tuple2<Integer, Integer>, Tuple2<Integer,Integer>> {

    private static final Logger logger = LoggerFactory.getLogger(ReducingStateFlatMap.class);

    // 创建ReducingState类型的状态结构
    private transient ReducingState<Tuple2<Integer,Integer>> reducingState;

    @Override
    public void flatMap(Tuple2<Integer, Integer> input, Collector<Tuple2<Integer, Integer>> collector) throws Exception {
        //将当前输入的元素添加到reducingState中，此时会调用ReducingStateDescriptor中定义的ReduceFunction函数进行聚合运算
        reducingState.add(input);
        // 获取状态中的聚合结果
        collector.collect(reducingState.get());
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        logger.info("{},{}", Thread.currentThread().getName(), "恢复或初始化状态");

        // 创建一个ReducingStateDescriptor状态描述符，指定名称为ReducingStateFlatMap，类型为 Tuple2<Integer, Integer>
        ReducingStateDescriptor<Tuple2<Integer,Integer>> descriptor = new ReducingStateDescriptor<>(
                "ReducingStateFlatMap",
                new ReduceFunction<Tuple2<Integer, Integer>>() {
                    @Override
                    public Tuple2<Integer, Integer> reduce(Tuple2<Integer, Integer> value01, Tuple2<Integer, Integer> value02) throws Exception {
                        return new Tuple2<>(value01.f0, value01.f1 + value02.f1);
                    }
                },
                TypeInformation.of(new TypeHint<Tuple2<Integer, Integer>>() {
                }));
        // 初始化reducingState
        reducingState = getRuntimeContext().getReducingState(descriptor);
        logger.info("初始化 reducingState ： {}", reducingState);
    }

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 封装了数据源函数，以及keyBy操作的数据流
        KeyedStream<Tuple2<Integer, Integer>, Integer> keyedStream = KeyStateBase.before(env);

        // 在keyedStream中使用有状态的FlatMap操作符
        DataStream<Tuple2<Integer, Integer>> resultStream = keyedStream.flatMap(new ReducingStateFlatMap());

        resultStream.print("输出结果 : ");

        env.execute("ReducingStateFlatMap");
    }
}
